#include <chrono>
#include <fstream>
#include <functional>
#include <iostream>
#include <memory>
#include <mutex>
#include <thread>
#include <vector>

#include "httplib.h"
#include "readerwritercircularbuffer.h"

struct block_t {
  std::shared_ptr<char> data_buf;
  uint32_t data_size = 0;

  block_t() = default;
  block_t(std::shared_ptr<char> data, uint32_t sz) : data_buf(data), data_size(sz) {
  }
};

using RingQueue = moodycamel::BlockingReaderWriterCircularBuffer<block_t>;
RingQueue ring_buffer(10);

// define callback function
using datacallback_t = std::function<void(std::shared_ptr<char>, uint32_t)>;
datacallback_t data_callback_ = nullptr;
std::mutex data_callback_mtx_;

void generate_point_cloud() {
  std::string pcd_file_path = "/home/dgliu/test/largefile_srv2/CBS.log";
  std::ifstream pcd_file(pcd_file_path, std::ios::binary);
  if (!pcd_file.is_open()) {
    std::cerr << "Error opening PCD file: " << pcd_file_path << std::endl;
    return;
  }

  pcd_file.seekg(0, std::ios::end);
  std::streampos file_size = pcd_file.tellg();
  pcd_file.seekg(0, std::ios::beg);

  const size_t buffer_size = 81920;

  while (true) {
    std::shared_ptr<char> buffer(new char[buffer_size], std::default_delete<char[]>());

    pcd_file.read(reinterpret_cast<char*>(buffer.get()), buffer_size);
    std::streamsize bytes_read = pcd_file.gcount();

    if (bytes_read <= 0) {
      pcd_file.clear();
      pcd_file.seekg(0, std::ios::beg);
      pcd_file.read(reinterpret_cast<char*>(buffer.get()), buffer_size);
      bytes_read = pcd_file.gcount();
    }

    data_callback_mtx_.lock();
    // do callback function here
    if (data_callback_ != nullptr) {
      data_callback_(buffer, static_cast<uint32_t>(bytes_read));
    }
    data_callback_mtx_.unlock();
    std::this_thread::sleep_for(std::chrono::microseconds(100));  // 100us
  }

  pcd_file.close();
}

void on_data_callback(std::shared_ptr<char> data, uint32_t size) {
  block_t en_block{data, size};
  bool en_ret = ring_buffer.try_enqueue(en_block);
  while (!en_ret) {
    std::this_thread::sleep_for(std::chrono::milliseconds(10));  // 10ms
    std::cerr << "sleep_for milliseconds(10)" << std::endl;
    en_ret = ring_buffer.try_enqueue(en_block);
  }
}

int main(void) {
  // always generate data
  std::thread generator_thread(generate_point_cloud);
  generator_thread.detach();

  httplib::Server svr;
  svr.Get("/largefile", [](const httplib::Request&, httplib::Response& res) {
    size_t total_size = 100 * 1024 * 1024;  // We will send 100 MB of data
    size_t sent_size = 0;
    data_callback_ = on_data_callback;

    res.set_chunked_content_provider(
        "application/octet-stream", [sent_size, total_size](size_t length, httplib::DataSink& sink) mutable {
          while (sent_size < total_size) {
            block_t de_block;
            ring_buffer.wait_dequeue(de_block);

            /*while (!ring_buffer.wait_dequeue_timed(de_block, std::chrono::milliseconds(1))) {
              std::cout << "timeout" << std::endl;
            }*/
            sink.write(reinterpret_cast<const char*>(de_block.data_buf.get()), static_cast<size_t>(de_block.data_size));
            sent_size += de_block.data_size;
            // std::cout << "[2] ring_buffer.size(): " << ring_buffer.size_approx() << std::endl;
          }
          data_callback_mtx_.lock();
          data_callback_ = nullptr;
          data_callback_mtx_.unlock();
          sink.done();

          return true;
        });

    res.set_header("Content-Disposition", "attachment; filename=largefile.dat");
  });

  svr.listen("localhost", 8080);

  return 0;
}
